#pragma once
#ifndef __linux__
#error unix local stream server only support linux system
#endif

#include <string>
#include <functional>
#include <memory>				//shared_ptr use
#include <queue>
#ifndef ASIO_STANDALONE
#define ASIO_STANDALONE
#define ASIO_NO_DEPRECATED
#endif
#include "asio/io_context.hpp" 	//io_context use
#include "asio/steady_timer.hpp"
#include "asio/local/stream_protocol.hpp"		//acceptor and socket use
#include "asio/read.hpp"
#include "asio/write.hpp"
#include "asiobuf.h"
#include "packet.h"

namespace KRPC {

using SendFunc = std::function<void(Packet &&)>;
using RecvCallback = std::function<bool(Packet &, const SendFunc &)>;
using ConnectCallback = std::function<void(const SendFunc &)>;
using CloseCallback = std::function<void()>;

class LOCALSTREAM_ASIO
{
private:
	void setSrcAddr()
	{
		char softlink[40] = {0}, exePath[128] = {0};
		sprintf(softlink, "/proc/%d/exe", getpid());
		int count = readlink(softlink, exePath, sizeof(exePath));
		assert(count > 0 && count <= (int)(sizeof(exePath)));
		const char *name = strrchr(exePath, '/');
		name = (name == nullptr) ? exePath : (name + 1);
		//前11个字节填'/tmp/local/',中间填name,后面17个字节填'_6位整数pid_stream4位整数sockfd',最后1字节填'\0'
		struct stat st;
		if (stat("/tmp/local", &st) == -1 || S_ISDIR(st.st_mode) == 0) {
			(void) mkdir("/tmp/local", 0777);	//析构函数中增加了删除逻辑，不需要依赖其他进程去清除
		}
		// tmp 目录重启后会清空 进程名+pid+协议类型+socketfd , 在同一系统同一时刻具有唯一性
		char path[128] = {0};
		int n = sprintf(path, "/tmp/local/%s_%d_stream%d", name, getpid(), m_socket.native_handle());
		path[n] = '\0';
		if (access(path, F_OK) == 0)
			unlink(path);
		srcPath = path;
	}
public:
	LOCALSTREAM_ASIO(asio::io_context &ioctx, const char *filename) : m_socket(ioctx), timer(ioctx), dstEndpoint(std::string("/tmp/local/") + filename)
	{
		asio::error_code ec;
		m_socket.open();
		setSrcAddr();
		m_socket.bind(asio::local::stream_protocol::endpoint(srcPath), ec);
		stopped = false;
	}

	~LOCALSTREAM_ASIO()
	{
		stop();
	}

	void start(RecvCallback recvCallback, uint32_t msec = 0, ConnectCallback connectCallback = nullptr, CloseCallback closeCallback = nullptr)
	{
		if (stopped)
			stopped = false;
		onRecv = std::move(recvCallback);
		onConnect = std::move(connectCallback);
		onClose = std::move(closeCallback);
		timeout = std::chrono::milliseconds(msec);
		sendFunc = [&](Packet &&packet){
			std::shared_ptr<KRPC::Buffer> buffer = std::make_shared<KRPC::Buffer>(packet.release());
			if (!buffer || !(*buffer))
				return;
			sndque.emplace(std::move(buffer));
			if (sndque.size() == 1)
				async_send();
		};
		asyncConnect();
	}

	void stop()
	{
		if (!stopped) {
			stopped = true;
			close();
		}
	}

private:
	asio::local::stream_protocol::socket m_socket;
	asio::steady_timer timer;
	asio::local::stream_protocol::endpoint dstEndpoint;
	std::string srcPath;
	std::queue<std::shared_ptr<KRPC::Buffer>> sndque;
	MsgHead head;
	RecvCallback onRecv;
	ConnectCallback onConnect;
	CloseCallback onClose;
	SendFunc sendFunc;
	std::chrono::milliseconds timeout;
	bool stopped;

	void close()
	{
		if (!m_socket.is_open())
			return;
		asio::error_code ec;
		m_socket.close(ec);
		timer.cancel();
		unlink(srcPath.c_str());
		if (onClose)
			onClose();
	}

	void timing()
	{
		if (timeout.count() > 0) {
			timer.expires_after(timeout);
			timer.async_wait([this](const std::error_code &ec){
				if (stopped)
					return;
				if (!ec) {	//说明定时器到期
					reconnect();
				}
			});
		}
	}

	void asyncConnect()
	{
		timer.expires_after(std::chrono::milliseconds(500));	//连接失败时每隔500ms重连
		timer.async_wait([this](const std::error_code &ec){
			if (stopped)
				return;
			if (!ec) {	//说明定时器到期
				asyncConnect();
			}
		});
		m_socket.async_connect(dstEndpoint, [this](const asio::error_code &ec){
			if (!ec) {
				timer.cancel();
				async_recv_head();
				if (onConnect)
					onConnect(sendFunc);
			}
		});
	}

	void reconnect()
	{
		if (stopped)
			return;
		close();
		if (m_socket.is_open())
			return;
		m_socket = asio::local::stream_protocol::socket(m_socket.get_executor());
		m_socket.open();
		setSrcAddr();
		asio::error_code ec;
		m_socket.bind(asio::local::stream_protocol::endpoint(srcPath), ec);
		asyncConnect();
	}

	//异步发送
	void async_send()
	{
		if (sndque.empty() || stopped)
			return;
		std::shared_ptr<KRPC::Buffer> buffer = sndque.front();
		auto callback = [this](asio::error_code ec, std::size_t n){
			if (!ec) {
				sndque.pop();
				async_send();
			} else {
				reconnect();
			}
		};
		asio::async_write(m_socket, asio::SendBuffer(buffer), callback);
	}

	//异步接收消息头
	void async_recv_head()
	{
		if (stopped)
			return;
		auto callback = [this](asio::error_code ec, std::size_t n){
			if (!ec && n == sizeof(head) && head.len >= sizeof(MsgHead)) {
				timer.cancel();
				size_t bodysize = head.len - sizeof(MsgHead);
				if (bodysize > 0) {
					asio::RecvBuffer rcvbuf(bodysize + sizeof(head), &head, sizeof(head));
					if (rcvbuf) {
						async_recv_body(rcvbuf);
						return;
					}
				} else if (bodysize == 0) {	//说明本次数据包只有头部没有数据体
					Packet packet(sizeof(MsgHead), sizeof(MsgHead));
					std::memcpy(packet.getHead(), &head, sizeof(MsgHead));
					if (onRecv(packet, sendFunc)) {
						async_recv_head();
						return;
					} 
				}
			}
			reconnect();
		};
		timing();
		asio::async_read(m_socket, asio::buffer(&head, sizeof(head)), callback);
	}
	//异步接收消息体
	void async_recv_body(asio::RecvBuffer rcvbuf)
	{	
		if (stopped)
			return;
		auto callback = [this, rcvbuf](asio::error_code ec, std::size_t n){
			if (!ec) {
				timer.cancel();
				Packet packet(std::move(*(rcvbuf.buffer())));
				if (packet.valid() && onRecv(packet, sendFunc)) {
					async_recv_head();
					return;
				}
			}
			reconnect();
		};
		timing();
		asio::async_read(m_socket, rcvbuf, callback);
	}
};

}
